package org.kth.dks.dks_node;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.Vector;

import org.apache.log4j.Logger;
import org.kth.dks.DKSAppInterface;
import org.kth.dks.DKSCallbackInterface;
import org.kth.dks.DKSMessagingInterface;
import org.kth.dks.DKSObject;
import org.kth.dks.DKSObjectTypes;
import org.kth.dks.dks_comm.ConnectionManager;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_comm.ThreadPool;
import org.kth.dks.dks_exceptions.DKSIdentifierAlreadyTaken;
import org.kth.dks.dks_exceptions.DKSNodeAlreadyRegistered;
import org.kth.dks.dks_exceptions.DKSRefNoResponse;
import org.kth.dks.dks_exceptions.DKSTooManyRestartJoins;
import org.kth.dks.dks_marshal.AckByeMsg;
import org.kth.dks.dks_marshal.AckHelloMsg;
import org.kth.dks.dks_marshal.AckJoinInitMsg;
import org.kth.dks.dks_marshal.AdaptLeaveMsg;
import org.kth.dks.dks_marshal.AdaptMsg;
import org.kth.dks.dks_marshal.BadPointerMsg;
import org.kth.dks.dks_marshal.BecomeNormalMsg;
import org.kth.dks.dks_marshal.BroadCastMsg;
import org.kth.dks.dks_marshal.ByeMsg;
import org.kth.dks.dks_marshal.CorrectionOnChangeInterface;
import org.kth.dks.dks_marshal.CorrectionOnJoinMsg;
import org.kth.dks.dks_marshal.CorrectionOnLeaveMsg;
import org.kth.dks.dks_marshal.DKSMarshal;
import org.kth.dks.dks_marshal.DKSMessage;
import org.kth.dks.dks_marshal.FailureMsg;
import org.kth.dks.dks_marshal.ForwardedMsgInterface;
import org.kth.dks.dks_marshal.HeartbeatMsg;
import org.kth.dks.dks_marshal.HelloMsg;
import org.kth.dks.dks_marshal.InsertNodeMsg;
import org.kth.dks.dks_marshal.JoinInitMsg;
import org.kth.dks.dks_marshal.LeaveAcceptMsg;
import org.kth.dks.dks_marshal.LeaveRejectMsg;
import org.kth.dks.dks_marshal.LeaveRequestMsg;
import org.kth.dks.dks_marshal.LookupMsg;
import org.kth.dks.dks_marshal.LookupRequestMsg;
import org.kth.dks.dks_marshal.LookupResponseMsg;
import org.kth.dks.dks_marshal.LookupResultMsg;
import org.kth.dks.dks_marshal.NodeLeftMsg;
import org.kth.dks.dks_marshal.RestartJoinMsg;
import org.kth.dks.dks_marshal.RestartOperationMsg;
import org.kth.dks.util.AsyncOperation;
import org.kth.dks.util.CommunicationInfo;
import org.kth.dks.util.DKSPrintTypes;
import org.kth.dks.util.MathMisc;
import org.kth.dks.util.MathMiscConstant;
import org.kth.dks.util.MessageInfo;
import org.kth.dks.util.NodeInfo;
import org.kth.dks.util.OperationType;
import org.kth.dks.util.Pair;
import org.kth.dks.util.RTEntry;

public class DKSNode
    implements DKSCallbackInterface, DKSMessagingInterface {
  
    private static Logger log = Logger.getLogger(DKSNode.class);  

    public DKSMarshal marshal = null;
	public static long N = 4294967296L; // 2^32 // 2147483647; //1073741824;
	public static int K = 16;
	public static int L = 8; // INVARIANT: this.k^^this.L = this.N
	public static int F = 6;

	private final static int MAX_TIMEOUTS = 5;
	private final static int TIMEOUT_WINDOW = 10;
	private int numTimeouts = 0;
	private long lastTimeoutId = 0;
	
  Vector jWSet = new Vector();
  Vector lWSet = new Vector();

  final static int RandomWaitAndRestart = 0;
  final static int LeaveAnywayBye = 1;
  int whatToDoWithLeaveReject = LeaveAnywayBye;

  DKSRef inserter = null;
  DKSRef predecessor = null;

  DKSRef successor = null;
  DKSStatus status = DKSStatus.NILL;
  boolean requestingToLeave = false;
  DKSRef pP = null;
  DKSRef pS = null;

  boolean IamInsertingNode = false;
  boolean IamRemovingNode = false;
  boolean stateReceived = false;

  InsertManager insertManager = new InsertManager();
  AsyncOperation joinFuture = null;
  AsyncOperation leaveFuture = null;
  DKSAppInterface appHandler = new DefaultAppHandler(this);

  protected ConnectionManager cm = null;

  RoutingTree routingTable = null;
  OrderedDKSRefList frontList = null;
  OrderedDKSRefList backList = null;

  LinkedList statisticsSentMessages = new LinkedList();

  DKSRef myDKSRef = null;

  MathMiscConstant math = new MathMiscConstant(DKSNode.N, DKSNode.K);

  private Nonces removedNonces = new Nonces();

  private static final int NONCESENDSIZE = 100;
  
  private ThreadPool threadPool = null;

//for test
  boolean IamPrinting = false;
  String data = null;

  private int MAX_RESTARTS = 25; // MAXIMUM NUMBER OF RETRIES WHEN JOINING
  private int joinExpBackoff = 1;
  private int joinBackoffCount = 0;
  
  private class Heartbeat extends TimerTask {
	  private long counter = 0;
	  public void run() {
		  if (predecessor==null || predecessor.equals(myDKSRef)) return;
		  HeartbeatMsg h = new HeartbeatMsg(counter++);
		  if (!send(predecessor, h)) {
			  failureHandler(predecessor, new FailureMsg(myDKSRef, predecessor, h));
		  }
	  }
  }

  private Heartbeat heartbeat = new Heartbeat();
  private final Timer heartbeatTimer = new Timer(DKSNode.class.getName()
		  + ".HeartbeatTimer");

  public DKSNode() {
  }

  public void setConnectionManager(ConnectionManager cm, DKSRef _dksRef) throws DKSNodeAlreadyRegistered
  {
    myDKSRef = _dksRef;
    this.cm = cm;
    marshal = cm.getDKSMarshal();

    frontList = new OrderedDKSRefList(myDKSRef, F, N, true);
    backList = new OrderedDKSRefList(myDKSRef, F, N, false);

    jWSet = new Vector();
    lWSet = new Vector();
    routingTable = new RoutingTree(N, K, L, myDKSRef);

    cm.registerDKSNode(myDKSRef, this);
    installHandlers(); // INVARIANT: has to be called *AFTER* the marshal is set
    
    threadPool = ThreadPool.getInstance();

    long intDel = (long)Math.max(1000, cm.getNodeRTT(predecessor));
    intDel = Math.min(5000, intDel); 
    heartbeatTimer.schedule(heartbeat, 0, intDel);
  }

  public void setMarshall(DKSMarshal dksMarshall) {
    marshal = dksMarshall;
  }

  private synchronized void updateMsgStatistics(DKSRef r, String s) {
    statisticsSentMessages.addLast(new MessageInfo(r,
        myDKSRef,
        s));
  }

  public boolean send(DKSRef dest, DKSMessage msg) {
    log.info("DKSNode:" + msg.getName() +
                     ":" + myDKSRef.getID() + "==>" + dest.getID());
    updateMsgStatistics(dest, msg.getName());
    return marshal.send(myDKSRef, dest, msg);
  }

  private void updateRingInfo(DKSRef[] d) {
    for (int i = 0; i < d.length; i++) {
      updateRingInfo(d[i]);
    }
  }

  private void updateRingInfo(DKSRef x) {
    // Check to see if we know that the node x
    // is lost.
    if (removedNonces.containsNonce(x)) {
      log.warn( "Node not added to RT because it has an old nonce [dksref="+x+"]");
      return;
    }

    if (x.getID() != myDKSRef.getID()) {
      log.debug( "Incoporating node " + x.getID());
      routingTable.foundNewNode(x);
      frontList.add(x);
      backList.add(x);
      log.debug(
                       "LeafSet " + backList.toStringReverse() + "--> " +
                       myDKSRef.getID() + " --> " + frontList);
    }
  }


  public void nodeLeftH(DKSRef coord, NodeLeftMsg m) {
    boolean removed = nodeFailed(m.getOldNode());
    updateRingInfo(coord);
    for (Iterator it=m.getLiveRefs().iterator(); it.hasNext();) {
      DKSRef curr = (DKSRef)it.next();
      updateRingInfo(curr);
    }

    if (coord.equals(successor) && removed) {
      send(coord, new NodeLeftMsg(m.getOldNode(), backList, removedNonces.getRecent(NONCESENDSIZE)));
    }

    for (Iterator it=m.getNonces().iterator(); it.hasNext();) {
      DKSRef curr = (DKSRef)it.next();
      nodeLeft(curr);
    }
  }

  private boolean nodeFailed(DKSRef failedNode) {
    nodeLeft(failedNode);
    boolean removed = false;

    if (failedNode.equals(successor)) {
      removed = true;
      if (frontList.size()<1) {
        log.error(
                         "*MAJOR FAILURE* frontlist exhausted, cannot find replacement for failed successor (DKS F parameter too low?)");
        successor = myDKSRef;
        predecessor = myDKSRef;
      }
      else
        successor = (DKSRef) frontList.get(0);
    }

    if (failedNode.equals(predecessor)) {
      removed = true;
      if (backList.size()<1)
        log.error(
                         "*MAJOR FAILURE* backlist exhausted, cannot find replacement for failed predecessor (DKS F parameter too low?)");
      else
        predecessor = (DKSRef) backList.get(0);
    }
    return removed;
  }

  private void nodeLeft(DKSRef leftNode) {

    removedNonces.addNonce(leftNode);  // make sure it wont pop back

    routingTable.nodeLost(leftNode);
    frontList.remove(leftNode);
    backList.remove(leftNode);
  }

  public Set getNeighboursSet() {
	  Set set = new HashSet();
	  set.addAll(frontList.toList());
	  set.addAll(backList.toList());
	  set.addAll(routingTable.toList());
	  return set;
  }
  
  private DKSRef[] getBackAndFrontList() {
    LinkedList ans = frontList.toList();
    LinkedList tmp = backList.toList();
    while (!tmp.isEmpty()) {
      DKSRef ele = (DKSRef) tmp.removeFirst();
      if (!ans.contains(ele)) {
        ans.add(ele);
      }
    }
    return (DKSRef[]) ans.toArray(new DKSRef[0]);
  }

  private DKSRef[] getKnownDKSRefs() {
    LinkedList ans = frontList.toList();
    LinkedList tmp = backList.toList();
    while (!tmp.isEmpty()) {
      DKSRef ele = (DKSRef) tmp.removeFirst();
      if (!ans.contains(ele)) {
        ans.add(ele);
      }
    }
    tmp = routingTable.toList();
    while (!tmp.isEmpty()) {
      DKSRef ele = (DKSRef) tmp.removeFirst();
      if (!ans.contains(ele)) {
        ans.add(ele);
      }
    }
    return (DKSRef[]) ans.toArray(new DKSRef[0]);
  }

//JOIN

  //To be called to join (nj is myself)
  public AsyncOperation newNode(DKSRef nj, DKSRef n) throws DKSRefNoResponse {
    if (nj.getID() < 0 || nj.getID() >= N) {
      log.error( "DKS ID is not valid ("+nj.getID()+")");
      return null;
    }
    pP = null;
    pS = null;
    whatToDoWithLeaveReject = RandomWaitAndRestart;
    IamInsertingNode = false;
    IamRemovingNode = false;
    stateReceived = false;
    joinFuture = AsyncOperation.start(OperationType.JOINTYPE);
    
    if (n == null) { //Join an empty DKS
      joinFuture.complete("FirstNode joined");
      predecessor = nj;
      successor = nj;
      status = DKSStatus.INSIDE;
      inserter = nj;
      
      log.debug("created a new dks");
    }
    else { //Join a non-empty DKS
      inserter = null;
      predecessor = null;
      successor = null;
      status = DKSStatus.GETTINGIN;
      pS = null;
      log.debug("about to join node " +nj.getID()+" with type ADDRESS and msgId="+joinFuture.getKey());
      log.debug( "Joining node "+nj.getID()+" with type ADDRESS and msgId="+joinFuture.getKey());
      LookupRequestMsg lm = new LookupRequestMsg(nj.getID(), LookupType.ADDRESS,
                                                 joinFuture.getKey(), true);
      if (!send(n, lm)) {
        heartbeatTimer.cancel();
        throw new DKSRefNoResponse(n.getDKSNetAddress());
      }
    }
    return joinFuture;
  } //newNode

  //nj wants to join us!

  public void insertNodeH(DKSRef nj, InsertNodeMsg msg) {
    log.debug(
                     "DKSNode:INSERTNODE:" + nj.getID() + "-->" +
                     myDKSRef.getID());
    if (status==DKSStatus.INSIDE) {
        if (MathMisc.belongsTonn(nj.getID(), predecessor.getID(),
                                 myDKSRef.getID(), N) &&
            pP == null && !requestingToLeave) {
          pP = nj;
          IamInsertingNode = true;
          log.debug(
                           "Sending front and back-list to " + nj);
          log.debug(
                           backList.toString() + frontList.toString());
          send(nj, new JoinInitMsg(predecessor, getBackAndFrontList())); //new marshal
        }
        else {
          log.debug(
                           "RestartJoin sent; reason: nj=" + nj.getID() + " p=" +
                           predecessor.getID() +
                           " n=" + myDKSRef.getID() + " pP=" +
                           Boolean.toString(pP == null) +
                           " reqLeave=" + requestingToLeave);
          send(nj, new RestartJoinMsg());
        }
      } else if (status==DKSStatus.GETTINGIN) {

        //XMLMessage restartJoinMsg = marshall.restartJoinMsg();
        //send(nj, restartJoinMsg);
        log.debug(
                         "RestartJoin sent; reason: gettingIn");
        send(nj, new RestartJoinMsg());
      } else if (status == DKSStatus.GETTINGOUT) {
        log.debug(
                         "RestartJoin sent; reason: gettingIn");
        send(nj, new RestartJoinMsg());
    }
  } //insertNodeH

  /**
   * Called when this node receives a message back because a node could not currently handle it
   * Waits a random interval and tries again
   * TODO: fix so that it does exponential backoff
   * @param from DKSRef
   * @param msg RestartJoinMsg
   */

  private Map restartOpBackoffMap = new HashMap();

  public void restartOperationH(DKSRef from, RestartOperationMsg msg) {
    DKSMessage operation = msg.getOperation();

    if (!restartOpBackoffMap.containsKey(from)) {
      restartOpBackoffMap.put(from, new Long(100));
    }
    long secToWait = ( (Long) restartOpBackoffMap.get(from)).longValue();

    if (Math.log( (double) secToWait) / Math.log(2) > MAX_RESTARTS) {
      secToWait = 1;
      log.error(
                       "Restarted operation " + operation.getName() +
                       " from node " + from + " MAX=" +
                       MAX_RESTARTS + " times (shock starting)");
    }

    secToWait += (new Random()).nextInt(100);
    try {
	Thread.sleep(secToWait);
    }
    catch (Exception ex) {
      log.error(
                       "Was interrupted while sleeping in restartOperationH\n" +
                       ex);
    }
    secToWait *= 2;
    restartOpBackoffMap.put(from, new Long(secToWait));

    send(from, operation); // resend the operation
  }

  public void joinInitH(DKSRef np, JoinInitMsg msg) {
    DKSRef rP = msg.getP();
    DKSRef[] dArray = msg.getDKSRefs();

    log.debug(
                     "DKSNode:JOININIT:" + np.getID() + "-->" +
                     myDKSRef.getID());
    if (status == DKSStatus.INSIDE) {
        log.debug(
                         "DKSNode:joinInitH:error:status is inside");
    } else if (status == DKSStatus.GETTINGIN ) {
        if (inserter == null && np.getID() == pS.getID()) {
          inserter = np;
          predecessor = rP;
          stateReceived = true;

          // dList is all DKSRefs known to the other node.
          // Here, the front and back-lists are filled, as well as
          updateRingInfo(np);
          updateRingInfo(rP);
          updateRingInfo(dArray);
          log.debug(
                           "JoinInit, not yet added " + np.getID());
          log.debug(
                           backList + " --> " + myDKSRef.getID() + " --> " +
                           frontList);

          final DKSRef _pred = predecessor;
          final DKSRef _succ = np;
          Runnable callbackObj = new Runnable() {
            public void run() {
              appHandler.joinCallback(_pred, _succ);
            }
          };
          //Thread thread = new Thread(callbackObj);
          //thread.setName(DKSNode.class.getName() + ".joinHandlerCallback");
          //thread.start(); // spawn callback in a separate thread
          threadPool.addJob(callbackObj);

        }
    } else if (status == DKSStatus.GETTINGOUT) {
        log.debug(
                         "DKSNode:joinInitH:error:status is gettingout");
    }
  } //joinInitH

  public void joinCallbackReturn() {
    log.debug(
                     "JoinCallbackReturn, from appllication level");
    if (status == DKSStatus.GETTINGOUT || status == DKSStatus.INSIDE) {
      log.error(
                     "Received a joinCallbackReturn while status!=GettingIn");
    }
    else {
      log.debug( "Sending on backlist " + backList);
      DKSRef[] bNfL = getBackAndFrontList();
      for (int i = 0; i < bNfL.length; i++) {
        if (bNfL[i].getID() != myDKSRef.getID()) {
          jWSet.add(bNfL[i]);
          log.debug(
                           "Sending hello to " + bNfL[i].getID() +
                           " JoinWait setsize: " + jWSet.size());
          send(bNfL[i], new HelloMsg("BACK_AND_FRONT_LIST")); //new marshal

        }
      }
    }
  }

  public void helloH(DKSRef nj, HelloMsg msg) {
    String typelist = msg.getType();

    log.debug( "DKSNode:HELLO:" + nj.getID() + "-->" +
                     myDKSRef.getID() + " type(" + typelist + ")");
    updateRingInfo(nj);
    send(nj, new AckHelloMsg(getBackAndFrontList())); //new marshal
  }

  public void ackHelloH(DKSRef nj, AckHelloMsg msg) {
    DKSRef[] dArray = msg.getDKSRefs();

    log.debug(
                     "DKSNode:ACKHELLO:" + nj.getID() + "-->" +
                     myDKSRef.getID() + " JoinWait setsize: " + jWSet.size());
    if (status == DKSStatus.NILL) {
    } else if (status == DKSStatus.GETTINGIN) {
        updateRingInfo(nj);
        updateRingInfo(dArray);

        jWSet.remove(nj);

        if (jWSet.size() == 0) {

          AckJoinInitMsg ajim = new AckJoinInitMsg(); //new marshal

          send(pS, ajim);
        }
    } else if (status == DKSStatus.INSIDE) {
        log.debug(
                         "DKSNode:ackHello:error:status is inside");
    } else if (status == DKSStatus.GETTINGOUT) {
        log.debug(
                         "DKSNode:askHello:error:status is gettingout");
    }
  } //ackHelloH

  public void ackJoinInitH(DKSRef nj, AckJoinInitMsg msg) {
    log.debug(
                     "DKSNode:ACKJOININIT:" + nj.getID() + "-->" +
                     myDKSRef.getID());
    if (status == DKSStatus.INSIDE || status == DKSStatus.GETTINGOUT) {
      if (IamInsertingNode && pP.getID() == nj.getID()) {
        DKSRef oldP = predecessor;
        predecessor = pP;
        updateRingInfo(pP);
        pP = null;
        IamInsertingNode = false;
        send(nj, new BecomeNormalMsg(oldP)); //new marshal
      }
    }
    else {
      log.debug(
                       "DKSNode:ackJoinInitH:error:status is wrong");
    }
  } //ackJoinInit

  private void initializeRT() {
	  for (int l = routingTable.a_levels; l >= 1; l--) {
		  for (int i = 1; i < routingTable.a_k_factor; i++) {
			  AsyncOperation find = AsyncOperation.start(OperationType.DATATYPE);
			  long start = routingTable.getBegin(l, i);
			  LookupRequestMsg lm = new LookupRequestMsg(start,
					  LookupType.JOINADDRESS,
					  find.getKey(), true);
			  send(myDKSRef, lm);
		  }
	  }
	  /*
	   AsyncOperation find = AsyncOperation.start(OperationType.DATATYPE);
	   long start = routingTable.getBegin(1, 1);
	   LookupRequestMsg lm = new LookupRequestMsg(start,
	   LookupType.JOINADDRESS,
	   find.getKey(), true);
	   send(myDKSRef, lm);
	   */
  }
  
  Pair getNodeIntervals(long id) {	  
	  long myId = this.myDKSRef.getID();
	  long nodeId = math.distanceClockWise(myId, id);
	  int logLevel = (int) (Math.log((double)nodeId)/Math.log((double)this.K));
	  int level = this.L - logLevel;
	  int interval = (int) (nodeId==0 ? 0 : 
		  (nodeId/((long) Math.pow((double)this.K, (double)logLevel))) );
	  return new Pair(new Integer(level), new Integer(interval));
  }
  
  public void fixNextPointer(DKSRef targetRef) {
	  Pair p = getNodeIntervals(targetRef.getID());
	  int level = ((Integer) p.first()).intValue();
	  int interval = ((Integer) p.second()).intValue();
//	  System.out.println("myId:"+myDKSRef.getID()+" Got target:"+targetRef.getID());
//	  System.out.println("myId:"+myDKSRef.getID()+" level:"+level+" interval:"+interval);
	  if (interval!=0) {
		  interval++;
		  if (interval==this.K) {
			  interval = 1;
			  level--;
			  if (level==0) return;
		  }
		  AsyncOperation find = AsyncOperation.start(OperationType.DATATYPE);
		  long start = math.nThroughKPowerL(level)*interval;
//		  System.out.println("myId:"+myDKSRef.getID()+" Going for l:"+level+" interval:"+interval);
		  LookupRequestMsg lm = new LookupRequestMsg(start,
				  LookupType.JOINADDRESS,
				  find.getKey(), true);
		  send(myDKSRef, lm);
	  }
  }

  public void becomeNormalH(DKSRef nj, BecomeNormalMsg msg) {
    joinExpBackoff = 1; // for sure
    DKSRef newp = msg.getP();

    log.debug(
                     "DKSNode:BECOMENORMAL:" + nj.getID() + "-->" +
                     myDKSRef.getID());
    if (status == DKSStatus.INSIDE) {
        log.error(
                         "DKSNode:becomeNormaltH:error:status is inside");
    } else if (status == DKSStatus.GETTINGIN) {
        if (inserter.getID() == nj.getID()) {
          status = DKSStatus.INSIDE;
          successor = pS;
          predecessor = newp;
          updateRingInfo(pS);
          pS = null;
          send(predecessor, new AdaptMsg());
          log.debug(
                           "DKSNode:becomeNormaltH:JOIN FINISHED!!!");

          initializeRT();

          correctionOnChangeJoin();

          joinFuture.complete("Join completed");
        }
    } else if ( status == DKSStatus.GETTINGOUT ) {
        log.error(
            "DKSNode:becomeNormaltH:error:status is gettingout");
    }
  } //becomeNormal

  public List getDependentIntervals(DKSRef pred, DKSRef succ) {
    List depIntervals = new LinkedList();
    for (int l = 1; l <= DKSNode.L; l++) {
      for (int i = 1; i < DKSNode.K; i++) {
        long start = math.modMinus(pred.getID(), math.nThroughKPowerL(l));
        long end = math.modMinus(succ.getID(), math.nThroughKPowerL(l));
        start = math.modPlus(start, 1); // adjusted to [start,end[
        end = math.modPlus(end, 1);
        depIntervals.add(new Interval(start, end));
      }
    }
    IntervalOptimizer opt = new IntervalOptimizer(math);
    List collapsedIntervals = opt.collapseIntervals(depIntervals);
    long myId = succ.getID();
    long predId = pred.getID();
    List finalIntervals = opt.removeInterval(collapsedIntervals,
                                             new Interval(predId, myId));
    return finalIntervals;
  }

  public void correctionOnChangeJoin() {
    List intervals = getDependentIntervals(predecessor, myDKSRef);
    for (ListIterator iter = intervals.listIterator(); iter.hasNext(); ) {
      Interval curr = (Interval) iter.next(); // intervals represented as ]start,end]
      byte[] msg = new CorrectionOnJoinMsg(myDKSRef, curr).flatten();
      routeAsync(curr.start, new DKSObject(msg), true);
    }
  }

  public void correctionOnChangeLeave(DKSRef leavingPred, DKSRef leavingNode, DKSRef leavingSucc) {
    List intervals = getDependentIntervals(leavingPred, leavingNode);
    for (ListIterator iter = intervals.listIterator(); iter.hasNext(); ) {
      Interval currInterval = (Interval) iter.next(); // intervals represented as ]start,end]
      byte[] msg = new CorrectionOnLeaveMsg(leavingNode, leavingSucc,
          currInterval).flatten();
      routeAsyncFrom(currInterval.start, new DKSObject(msg), true, successor);
    }
  }

  public void cocJoinNotificationH(CorrectionOnJoinMsg msg) {
    log.debug(
                     "CorrectionOnChange at " + myDKSRef + ", Node " +
                     msg.getNewNode() + " joined");
    updateRingInfo(msg.getNewNode());

    Interval i = msg.getInterval();
    if (pP != null) {
      CorrectionOnJoinMsg cocJoin = new CorrectionOnJoinMsg(msg.getNewNode(),
          new Interval(pP.getID(), pP.getID()));
      send(pP, cocJoin);
    }

  }

  public void cocJoinInterleavedH(DKSRef succ, CorrectionOnJoinMsg msg) {
    updateRingInfo(msg.getNewNode());
  }

  public void cocLeaveNotificationH(CorrectionOnLeaveMsg msg) {
    log.debug(
                     "CorrectionOnChange at " + myDKSRef + ", Node " +
                     msg.getOldNode() + " with successor " +
                     msg.getNewNode() + " left");
    nodeLeft(msg.getOldNode());
    updateRingInfo(msg.getNewNode());
  }

  public  void correctionOnChangeH(CorrectionOnChangeInterface cocMsg) {
    Interval i = cocMsg.getInterval();

    if (math.belongsToI(myDKSRef.getID(), i.start, i.end)) {
      if (cocMsg instanceof CorrectionOnJoinMsg) {
        CorrectionOnJoinMsg joinMsg = (CorrectionOnJoinMsg) cocMsg;
        updateRingInfo(joinMsg.getNewNode());
      } else if (cocMsg instanceof CorrectionOnLeaveMsg) {
        CorrectionOnLeaveMsg leaveMsg = (CorrectionOnLeaveMsg) cocMsg;
        nodeLeft(leaveMsg.getOldNode());
        updateRingInfo(leaveMsg.getNewNode());
      }
      DKSObject rawMsg = new DKSObject( ((DKSMessage) cocMsg).flatten());
      send(myDKSRef, new BroadCastMsg(i.start, i.end, rawMsg, true));
    }
  }

  public void restartJoinH(DKSRef nj, RestartJoinMsg msg) throws
      DKSTooManyRestartJoins {
      
    if (joinBackoffCount >= MAX_RESTARTS) {
      joinExpBackoff = 1;
      joinBackoffCount = 0;
      log.debug("RestartJoinH cancel restarts"); 
      joinFuture.cancel(new DKSTooManyRestartJoins(MAX_RESTARTS));
      
      return;
    }

    Random rnd = new Random();
    log.debug("RestartJoinH going to sleep " + joinExpBackoff); 
    try {
      Thread.sleep(rnd.nextInt(1000) + joinExpBackoff);
    }
    catch (Exception ex) {
      log.warn(
          "Got exception while sleeping inside restartJoin handler\n" + ex);
    }
    joinExpBackoff *= 2;
    joinBackoffCount++;

    log.debug(
                     "DKSNode:restartJoinH:" + nj.getID() + "-->" +
                     myDKSRef.getID() + " msgId="+joinFuture.getKey());
    LookupRequestMsg lm = new LookupRequestMsg(myDKSRef.getID(),
                                               LookupType.ADDRESS,
                                               joinFuture.getKey(), true);
    send(nj, lm);
  } //restartJoinH

  public void adaptH(DKSRef nj, AdaptMsg msg) {
    log.debug( "DKSNode:adaptH:" + nj.getID() + "-->" +
                     myDKSRef.getID());
    successor = nj;
  }

  public void adaptLeaveH(DKSRef nj, AdaptLeaveMsg msg) {
    log.debug(
                     "DKSNode:adaptLeaveH:" + nj.getID() + "-->" +
                     myDKSRef.getID());
    successor = nj;
  }

//LEAVE
  public AsyncOperation prepareForLeave() {
    if (status == DKSStatus.INSIDE) {
      leaveFuture = AsyncOperation.start(OperationType.LEAVETYPE);

      if (pP == null) {
        requestingToLeave = true;

        if (predecessor.equals(myDKSRef))
        {
          heartbeatTimer.cancel();
          leaveFuture.complete("Leave completed (last node)");
          return leaveFuture;
        }
        send(successor, new LeaveRequestMsg(predecessor)); //new marshall
      }

      return leaveFuture;

    } else if (status == DKSStatus.GETTINGIN) {

      leaveFuture.complete("Leave completed automatically");
      return leaveFuture;
      //todo:must be completed.
    } else if (status == DKSStatus.GETTINGOUT) {
      log.error("DKSLeave:prepareForLeave:error:status is gettingout");
      leaveFuture.complete("Leave completed");
      return leaveFuture;
    }
    leaveFuture.complete("Leave completed");
    return leaveFuture;
  } //prepareForLeave

  public void leaveRequestH(DKSRef source, LeaveRequestMsg msg) {
    log.debug(
                     "DKSLeave:leaveRequestH from " + source.getID());
    DKSRef pnl = msg.getP();
    if (status == DKSStatus.INSIDE) {
        if (source.getID() == predecessor.getID()) {
          if (requestingToLeave) {
            if (source.getID() < myDKSRef.getID()) {
              send(source, new LeaveRejectMsg()); //new marshall
            }
            else {
              pP = pnl;
              requestingToLeave = false;
              IamRemovingNode = true;
              send(source, new LeaveAcceptMsg()); //new marshall
            }
          }
          else {
            if (pP == null) {
              pP = pnl;
              IamRemovingNode = true;
              send(source, new LeaveAcceptMsg()); //new marshall
            }
            else {
              send(source, new LeaveRejectMsg()); //new marshall
            }
          }
        }
    } else if (status == DKSStatus.GETTINGIN) {
        log.error("DKSLeave:leaveRequestH:error:status is gettingin");
    } else if (status == DKSStatus.GETTINGOUT) {
        if (source.getID() == predecessor.getID()) {
          send(source, new LeaveRejectMsg()); //new marshall
        }
    }
  } //leaveRequestH

  public void leaveAcceptH(DKSRef source, LeaveAcceptMsg msg) {
    log.debug(
                     "DKSNode:leaveAccceptH:" + source.getID() + "-->" +
                     myDKSRef.getID());
    if ( status == DKSStatus.INSIDE ) {
        if (requestingToLeave && source.getID() == successor.getID()) {

          final DKSRef _pred = predecessor;
          final DKSRef _succ = source;
          Runnable callbackObj = new Runnable() {
            public void run() {
              appHandler.leaveCallback(_pred, _succ);
            }
          };
          //Thread thread = new Thread(callbackObj);
          //thread.setName(DKSNode.class.getName() + ".leaveHandlerCallback");
          //thread.start(); // spawn callback in a separate thread
          threadPool.addJob(callbackObj);
        }
    } else if ( status == DKSStatus.GETTINGIN ) {
        log.debug(
                         "DKSLeave:leaveAcceptH:error:status is gettingin");
    } else if ( status == DKSStatus.GETTINGOUT ) {
        log.error("DKSLeave:leaveAcceptH:error:status is gettingout");
    }
  } //leaveAcceptH

  public void leaveCallbackReturn() {
    if (status == DKSStatus.GETTINGOUT) {
      log.error(
                     "Received a leaveCallback while status=gettingOut!");
    }
    else {
      status = DKSStatus.GETTINGOUT;
      goodBye();
    }
  }

  public void leaveRejectH(DKSRef source, LeaveRejectMsg msg) {
    if (status == DKSStatus.INSIDE) {
      if (requestingToLeave && source.getID() == successor.getID()) {
        switch (whatToDoWithLeaveReject) {
          case RandomWaitAndRestart:

            //todo
            break;
          case LeaveAnywayBye:
            lWSet = new Vector();
            DKSRef[] da = getBackAndFrontList();
            for (int i = 0; i < da.length; i++) {
              if (!lWSet.contains(da[i])) {
                lWSet.addElement(da[i]);
              }
            }
            goodBye();
            //todo
            break;
        }
      }
    } else if ( status == DKSStatus.GETTINGIN ) {
      log.debug(
                       "DKSLeave:leaveRejectH:error:status is gettingin");
    } else if ( status == DKSStatus.GETTINGOUT) {
      log.error("DKSLeave:leaveRejectH:error:status is gettingout");
    }
  } //leaveRejectH

  public void byeH(DKSRef source, ByeMsg msg) {
    DKSRef[] dArray = msg.getDKSRefs();
    String typeList = msg.getTypeList();

    log.debug( "byeH:BYE:" + source.getID() + "-->" +
                     myDKSRef.getID());
    if (status == DKSStatus.INSIDE || status == DKSStatus.GETTINGIN) {
      if (typeList.equals("FL")) {
        if (source.getID() == predecessor.getID()) {
          predecessor = pP;
          pP = null;
          IamRemovingNode = false;
          if (predecessor.equals(myDKSRef))
            successor = myDKSRef;
        }
      }
      log.debug(
                       "Before Bye: backlist:" + backList.toStringReverse() + " myId:" +
                       myDKSRef.getID() + " frontlist:" + frontList);
      log.debug( "Removing node " + source.getID());
      nodeLeft(source);
      log.debug(
                       "After bye: backlist" + backList.toStringReverse() + " myId:" +
                       myDKSRef.getID() + " frontlist:" + frontList);
      updateRingInfo(dArray);
      log.debug(
                       "After bye sorted: backlist:" + backList.toStringReverse() + " myId:" +
                       myDKSRef.getID() + " frontlist:" + frontList);

      // The successor is explicitly handled, both as an entry in the
      // front list and as an explicit variable. This should be removed.
      if (source.equals(successor)) {  // do not distinguish between FL and BL when deleting successor
//      if (typeList.equals("BL") && source.getID() == successor.getID()) {
        log.debug( "Changing successor from "+successor+" to "+frontList.getFirst());
        successor = (DKSRef) frontList.getFirst();
      }
      send(source, new AckByeMsg()); //new marshall
    }
    else {
      if (!stateReceived) {
        log.error( "not in state received -- byeH");
      }
      else {
        nodeLeft(source);
        updateRingInfo(dArray);
        send(source, new AckByeMsg()); //new marshall
      }
    }
  } //byeH

  public void ackByeH(DKSRef source, AckByeMsg msg) {
    if ( status == DKSStatus.INSIDE ) {
      log.debug(
                       "DKSLeave:ackByeH:error:status is inside");
    } else if ( status == DKSStatus.GETTINGIN ) {
      log.debug(
                       "DKSLeave:ackByeH:error:status is gettingin");
    } else if ( status == DKSStatus.GETTINGOUT ) {
      lWSet.remove(source);
      if (lWSet.size() == 0) {
        log.debug(
                         "DKSLeave:ackByeH:Leave is FINISHED!");

        correctionOnChangeLeave(predecessor, myDKSRef, successor);
        try {
          Thread.sleep(200); // let the dispatcher send these messages (no guarantee)
        } catch (Exception ex) {}
        predecessor = successor = myDKSRef;
        heartbeatTimer.cancel();

        leaveFuture.complete("Leave completed");
        //	      cm.end();  // cannot kill the whole ConnectionManager here, as other DKSNode's might be running
      }
    }
  } //ackByeH

  public void goodBye() {
    lWSet = new Vector();
    DKSRef[] fl = (DKSRef[]) frontList.toArray();
    for (int i = 0; i < fl.length; i++) {
      send(fl[i], new ByeMsg("FL", getBackAndFrontList()));
      lWSet.add(fl[i]);
    }

    DKSRef[] bl = (DKSRef[])backList.toArray();
    for (int i = 0; i < bl.length; i++) {
      if (!frontList.contains(bl[i])) {
        send(bl[i], new ByeMsg("BL", getBackAndFrontList()));
        lWSet.add(bl[i]);
      }
    }
  }

  public void lookupRequestH(DKSRef sender, LookupRequestMsg lm) {
    final long t = lm.getTarget();
    LookupType type = lm.getType();
    String msgId = lm.getMsgId();
    final boolean internal = lm.getInternal();
    final DKSObject payload = lm.getPayload(); // will be null if lm.getType()!=LookupType.DATA
    // because some messages carry payload, some dont!

    log.debug(
                     "LOOKUPREQUEST:" + sender.getID() + "-->" +
                     myDKSRef.getID() + " type(" + type + ") msgID(" +
                     msgId + ") " + status);

    if (status == DKSStatus.GETTINGIN) {
      log.debug(
                       "DKSNode:lookupReqH:error:Status is gettingin (restarting)!");
      send(sender, new RestartOperationMsg(lm));
    }
    else {
      if (MathMisc.belongsTo(t, predecessor.getID(), myDKSRef.getID(), N)) {
        log.debug(
                         "lookup_req:type:" + type + " from " + sender.getID());
        if (type == LookupType.ADDRESS || type == LookupType.JOINADDRESS) {
          DKSRef n = myDKSRef;
          LookupResponseMsg lrm = new LookupResponseMsg(n, type, msgId);
          send(sender, lrm);
        }
        else if (type == LookupType.ASYNCDATA) {
          log.debug(
                           "lookupRequestH:type:" + type + " from " +
                           sender.getID());
          Runnable callbackObj = new Runnable() {
            public void run() {
              routeCallbackAsync(t, payload, internal);
            }
          };
          //Thread thread = new Thread(callbackObj);
          //thread.setName(DKSNode.class.getName() + ".lookupRequestHandlerCallback");
          //thread.start(); // spawn callback in a separate thread
          threadPool.addJob(callbackObj);
        }
        else if (type == LookupType.SYNCDATA) {
          log.debug(
                           "lookupRequestH:type:" + type + " from " +
                           sender.getID());
          DKSRef n = myDKSRef;
          DKSObject responsePayload = appHandler.routeCallback(t, payload);
          LookupResponseMsg lrm = new LookupResponseMsg(n, type,
              responsePayload, msgId);
          send(sender, lrm);
        }
        else {
          log.debug(
                           "LookupType unknown (DKSNode:lookupRequestH:type:" +
                           type + ")   from " + sender.getID());
        }
      }
      else {
        log.debug(
                         "lookupRequestH Forwarding lookup for key: " + t +
                         " sent from " + sender.getID());
        LookupMsg lookupmsg;
        if (type == LookupType.ASYNCDATA || type == LookupType.SYNCDATA) {
          lookupmsg = new LookupMsg(t, type, sender,
                                    myDKSRef, payload, msgId, lm.getInternal());
        }
        else {
          lookupmsg = new LookupMsg(t, type, sender,
                                    myDKSRef, msgId, lm.getInternal());
        }
        forward(lookupmsg);
      }
    }
  }

  public void lookupH(DKSRef np, LookupMsg msg) {
    final long target = msg.getTarget();
    final boolean internal = msg.getInternal();
    LookupType type = msg.getType();
    Vector stack = msg.getStack();
    String msgId = msg.getMsgId();
    final DKSObject payload = msg.getPayload();

    log.debug( "DKSNode:LOOKUP:" + np.getID() + "-->" +
                     myDKSRef.getID() + " msgID(" + msgId + "), target="
                     + target + ", pred=" + predecessor.getID() + ", TYPE="
                     + type + ", STACK=" + stack.size());
    
    
    if (correctionOnUse(np, msg)) {
      // The correction on use mechanism took care of the message and sent it
      // to the proper process.
      return;
    }

    log.debug("AFTER COU");
    
    
    if (MathMisc.belongsTo(target, predecessor.getID(), myDKSRef.getID(), N)) {
      if (type == LookupType.ADDRESS || type == LookupType.JOINADDRESS ||
          type == LookupType.SYNCDATA) {
        DKSRef n = myDKSRef;
        DKSRef dest = (DKSRef) stack.elementAt(0);
        stack.removeElementAt(0);
        LookupResultMsg lrmsg = null;
        if (type == LookupType.ADDRESS || type == LookupType.JOINADDRESS) {
          lrmsg = new LookupResultMsg(n, type, stack, msgId);
        }
        else if (type == LookupType.SYNCDATA) {
          DKSObject responsePayload = appHandler.routeCallback(target, payload);
          lrmsg = new LookupResultMsg(n, type, stack, responsePayload, msgId);
        }
        send(dest, lrmsg); //new marshall
      }
      else if (type == LookupType.ASYNCDATA) {
        Runnable callbackObj = new Runnable() {
          public void run() {
            routeCallbackAsync(target, payload, internal);
          }
        };
        //Thread thread = new Thread(callbackObj);
        //thread.setName(DKSNode.class.getName() + ".lookupHandlerCallback");
        //thread.start(); // spawn callback in a separate thread
        threadPool.addJob(callbackObj);
      }
      else {
        log.error(
                       "Unknown LookupType (type=" + type + ")");
      }
    }
    else {
      log.debug("FORWARDING MSG because: target=" + target + " ");
      msg.getStack().add(0, myDKSRef);
      forward(msg);
    }
  }

  public void lookupResponseH(DKSRef np, LookupResponseMsg lrm) {
    DKSRef targetRef = lrm.getTargetRef();
    LookupType type = lrm.getType();
    String msgId = lrm.getMsgId();
    DKSObject payload = lrm.getPayload();

    log.debug(
                     "DKSNode:LOOKUPRESPONSE:" + np.getID() + "-->" +
                     myDKSRef.getID() + " msgID(" + msgId + ")");
    if (status == DKSStatus.INSIDE || status == DKSStatus.GETTINGOUT) {
      if (type == LookupType.SYNCDATA) {
	  AsyncOperation thisOp = AsyncOperation.get(msgId);
	  if(thisOp!=null)
	      thisOp.complete(payload);
      }
      else if (type == LookupType.ADDRESS) {
        AsyncOperation thisOp = AsyncOperation.get(msgId);
	if(thisOp!=null)
	    thisOp.complete(targetRef);
      }
      else if (type == LookupType.JOINADDRESS) {
        updateRingInfo(targetRef);
//        fixNextPointer(targetRef);
      }
    }
    else {
      if (type == LookupType.ADDRESS) {
        AsyncOperation thisOp = AsyncOperation.get(msgId);
        if (thisOp != null &&  ( (OperationType) thisOp.getState()) == OperationType.JOINTYPE && targetRef.getID()==myDKSRef.getID()) {
          joinFuture.cancel(new DKSIdentifierAlreadyTaken());
        } else if ( ( (OperationType) thisOp.getState()) == OperationType.JOINTYPE) {
          pS = targetRef;
          send(targetRef, new InsertNodeMsg());
        } else if (type == LookupType.JOINADDRESS) {
          updateRingInfo(targetRef);
//          fixNextPointer(targetRef);
        }
        else {
          log.error(
                           "DKSNode::lookupResponseH: Lookup (TYPE=ADDR) received while not in the system");
        }
      }
    }
  } //lookupResponseH

  public void lookupResultH(DKSRef np, LookupResultMsg msg) { //todo:to be completed
    DKSRef t = msg.getT();
    LookupType type = msg.getType();
    Vector stack = msg.getStack();
    String msgId = msg.getMsgId();
    DKSObject payload = msg.getPayload();

    log.debug("DKSNode:LOOKUPRESULT:" + np.getID() + "-->" +
                     myDKSRef.getID() + " msgID(" + msgId + ")), t="
            + t + ", pred=" + predecessor.getID() + ", TYPE="
            + type + ", STACK=" + stack.size());
    
    
    if (type == LookupType.ADDRESS || type == LookupType.JOINADDRESS ||
        type == LookupType.SYNCDATA) {
      if (stack.size() == 1) {
        LookupResponseMsg lrm = null;
        if (type == LookupType.ADDRESS || type == LookupType.JOINADDRESS) {
          lrm = new LookupResponseMsg(t, type, msgId);
        }
        if (type == LookupType.SYNCDATA) {
          lrm = new LookupResponseMsg(t, type, payload, msgId);

        }
        DKSRef dest = (DKSRef) stack.elementAt(0);
        send(dest, lrm);
      }
      else if (stack.size() > 1) {
        DKSRef dest = (DKSRef) stack.elementAt(0);
        stack.removeElementAt(0);
        LookupResultMsg lrm = null;
        if (type == LookupType.ADDRESS || type == LookupType.JOINADDRESS) {
          lrm = new LookupResultMsg(t, type, stack, msgId);
        }
        if (type == LookupType.SYNCDATA) {
          lrm = new LookupResultMsg(t, type, stack, payload, msgId);

        }
        send(dest, lrm); //new marshall
      }
    }
  } //lookupResultH

  private void broadcastCallbackInternal(DKSObject payload, boolean internal) {
    if (!internal) {
      appHandler.broadcastCallback(payload);
    }
    else {
      DKSMessage msg = DKSMessage.unmarshal(payload.getData()); // inefficient, should be avoided
      if (msg instanceof CorrectionOnJoinMsg) {
        cocJoinNotificationH( (CorrectionOnJoinMsg) msg);
      }
      else if (msg instanceof CorrectionOnLeaveMsg) {
        cocLeaveNotificationH( (CorrectionOnLeaveMsg) msg);
      }
      else {
        log.error(
            "broadcastCallbackInternal got an unknown internal message");
      }
    }
  }

  public void broadCastH(DKSRef sender, BroadCastMsg msg) {
    long start = msg.getStart();
    long limit = msg.getLimit();
    long myId = myDKSRef.getID();
    DKSObject payload = msg.getPayload();
    log.debug(
                     "ReceivedBroadCast from: " + sender.getID() + " start: " +
                     start + " limit: " + limit);
    broadcastCallbackInternal(payload, msg.getInternal());

    if (!sender.equals(myDKSRef) &&
        // !sender.equals(predecessor) &&
        MathMisc.belongsToII(predecessor.getID(), start, myId, N)) {
      // INVARIANT, from erik: Ali, is this restricted broadcast correct?
      send(predecessor, new BroadCastMsg(start, myId, payload, msg.getInternal()));
      send(sender, new BadPointerMsg(predecessor));
    }
    for (int l = 1; l <= routingTable.getLevels(); l++) {
      for (int i = K - 1; i > 0; i--) {
        DKSRef responsible = routingTable.getResponsible(l, i);
        if (MathMisc.belongsTonn(responsible.getID(), myId, limit, N)) {
          long bc_start = routingTable.getBegin(l, i);
          send(responsible,
               new BroadCastMsg(bc_start, limit, payload, msg.getInternal()));

          limit = bc_start;
        }

      }
    }

  } // broadCastH

  public void badPointerH(DKSRef sender, BadPointerMsg msg) {
    DKSRef p = msg.getP();
    log.debug(
                     "DKSNode:BADPOINTER:" + sender.getID() + "-->" +
                     myDKSRef.getID() + " better ref " + p.getID());
    updateRingInfo(p);
  }

  public void forward(ForwardedMsgInterface msg) {
    long target = msg.getTarget();
    DKSRef nextHop = routingTable.getIntervalResp(target);
    long start = routingTable.getIntervalStart(target);
    msg.setIntervalStart(start);
    send(nextHop, (DKSMessage) msg);
  }

  public DKSRef bestCandidate(DKSRef node, long target) {
    DKSRef cand = myDKSRef; // just as a watchdog.
    ListIterator iter = backList.listIterator();
    while (iter.hasNext()) {
      DKSRef ele = (DKSRef) iter.next();
      if (MathMisc.belongsTo(target, node.getID(), ele.getID(), N)) {
        cand = ele;
      }
    }
    return cand;
  }

  public boolean correctionOnUse(DKSRef sender, ForwardedMsgInterface msg) {
    updateRingInfo(sender);
    long start = msg.getIntervalStart();
    long target = msg.getTarget();
    if (sender.equals(predecessor)) {
      return false;
    }
    if (MathMisc.belongsTo(start, sender.getID(), predecessor.getID(), N)) {
      log.debug(
                       "CorUse sender: " + sender.getID() + " target: " +
                       target + " start: " + start);
      log.debug( "Backlist " + backList);
      DKSRef cand = bestCandidate(sender, start);
      log.debug( "bestCandidate: " + cand.getID());
      send(sender, new BadPointerMsg(cand));
      if (MathMisc.belongsTo(target, sender.getID(), predecessor.getID(), N)) {
        send(cand, (DKSMessage) msg);
        return true;
      }
    }
    return false;
  }

  public AsyncOperation findResponsible(long identifier) {
    AsyncOperation find = AsyncOperation.start(OperationType.FINDTYPE);

    LookupRequestMsg lm = new LookupRequestMsg(identifier, LookupType.ADDRESS,
                                               find.getKey(), true);
    send(myDKSRef, lm);
    return find;
  }

  public AsyncOperation route(long identifier, DKSObject payload) {
    AsyncOperation find = AsyncOperation.start(OperationType.DATATYPE);

    LookupRequestMsg lm = new LookupRequestMsg(identifier, LookupType.SYNCDATA,
                                               payload, find.getKey(), false);
    send(myDKSRef, lm);
    return find;
  }

  public void routeAsync(long identifier, DKSObject payload) {
    routeAsync(identifier, payload, false);
  }

  public void routeAsync(long identifier, DKSObject payload, boolean internal) {
    routeAsyncFrom(identifier, payload, internal, myDKSRef);
  }

  public void routeAsyncFrom(long identifier, DKSObject payload, boolean internal, DKSRef firstNode) {
    AsyncOperation find = AsyncOperation.start(OperationType.DATATYPE);

    LookupRequestMsg lm = new LookupRequestMsg(identifier, LookupType.ASYNCDATA,
                                               payload, find.getKey(), internal);
    send(firstNode, lm);
  }

  // INVARIANT: runs Asynchronously, and must make a call to appHandler if the msg is not handled
  private void routeCallbackAsync(long target, DKSObject payload,
                                  boolean internal) {
    if (payload.getType()==DKSObjectTypes.DKSRESTBCAST) {
      broadCastRestrictedCallback(payload);
    }
    else if (!internal) {
      appHandler.routeCallbackAsync(target, payload);
    }
    else {
      DKSMessage msg = DKSMessage.unmarshal(payload.getData()); // inefficient, should be avoided
      if (msg instanceof CorrectionOnChangeInterface) {
        correctionOnChangeH( (CorrectionOnChangeInterface) msg);
      }
      else {
        log.error(
                         "routeCallbackAsync got an unknown internal message");
      }
    }
  }

  public DKSMarshal getDKSMarshal() {
    return marshal;
  }

  public DKSCallbackInterface setCallbackHandler(DKSAppInterface appHandler) {
    this.appHandler = appHandler;
    return this;
  }

  public void search(long key, javax.swing.JTextField jtvalue, String msgId) {
  }

  public DKSObject[] lookup(long key) {
    return DKSLookup.lookup(this, key);
  }

  // These method are solely for monitoring and debug purposes

  public NodeInfo getNodeInfo() {
    int levels = routingTable.getLevels();
    NodeInfo ans = new NodeInfo(N, K,
                                //(int) (Math.log( (double) N) / Math.log( (double) K)));
				L, status);
    ans.frontList = (DKSRef[])frontList.toArray();
    ans.backList = (DKSRef[])backList.toArray();
    ans.predecessor = predecessor;
    ans.successor = successor;
    for (int l = 1; l <= levels; l++) {
      for (int i = 0; i < K; i++) {
        RTEntry rt = new RTEntry(routingTable.getBegin(l, i),
                                 routingTable.getEnd(l, i),
                                 routingTable.getResponsible(l, i));
        ans.routingTable[l - 1][i] = rt;
      }
    }
    return ans;
  }

  public java.io.Serializable getDebugInfo() {
    return null; // Ali: this was for P2PD3 project, can be removed.
  }

  public CommunicationInfo getComInfo() {
    return cm.getComInfo();
  }

  public synchronized MessageInfo[] getMessageInfo() {
    MessageInfo[] array = (MessageInfo[]) statisticsSentMessages.toArray(new MessageInfo[0]);
    statisticsSentMessages.clear();
    return array;
  }

  public DKSRef getDKSRef() {
    return myDKSRef;
  }

  public void broadCast(DKSObject message) {
    log.debug( "broadCast request");
    send(myDKSRef, new BroadCastMsg(myDKSRef.getID(), myDKSRef.getID(), message));
  }

  public void broadCastRestricted(DKSObject message, long startId, long endId) { // [ ]
    log.debug( "broadCastRestricted request");
    BroadCastMsg m = new BroadCastMsg(startId, endId, message);
    DKSObject obj = new DKSObject(DKSObjectTypes.DKSRESTBCAST, m.flatten());
    routeAsync(startId, obj);
//    send(myDKSRef, new BroadCastMsg(myDKSRef.getID(), myDKSRef.getID(), message));
  }

  public void broadCastRestrictedCallback(DKSObject message) {
    if (message.getType()==DKSObjectTypes.DKSRESTBCAST) {
      BroadCastMsg m = (BroadCastMsg) DKSMessage.unmarshal(message.getData());
      log.debug( "broadCastRest started from start of interval");
      send(myDKSRef, m);
    } else {
      log.error( "Got a broadCastRestrictedCallback() but message type was not DKSRESTBCAST");
    }
  }


  public void failureHandler(DKSRef src, FailureMsg fm) {
	  log.warn( "Failure sending from:"+fm.getSrc()+" to:"+fm.getDest()+" msg:"+fm.getMsg());
	  if (fm.getMsg() instanceof HeartbeatMsg) {
		  long currTimeoutId = ((HeartbeatMsg)fm.getMsg()).getValue();

		  if (Math.abs(currTimeoutId - lastTimeoutId)>TIMEOUT_WINDOW)
			  numTimeouts=0;
		  
		  if (numTimeouts++>=MAX_TIMEOUTS) {
			  numTimeouts=0;
			  final DKSRef failed = predecessor;
			  nodeLeft(failed);
			  predecessor = (DKSRef)backList.getFirst();
			  if (predecessor==null) predecessor=myDKSRef;
			  
			  nodeFailed(failed);

			  if (IamInsertingNode || IamRemovingNode || status!=DKSStatus.INSIDE)
				  log.warn("Node failure while doing atomic action: IamInserting:"+IamInsertingNode+" IamRemoving:"+IamRemovingNode+" failed:"+src+" status:"+status);
			  IamInsertingNode = false;
			  IamRemovingNode = false;
			  
			  
			  DKSRef pred = null;
			  
			  if (backList.size()>=1)
				  pred = (DKSRef)backList.get(0);
			  
			  if (pred!=null && failed!=null && myDKSRef!=null) {
//				  System.out.println("Doing COC leave on \n\tpred:"+pred+"\n\tfail:"+failed+"\n\tsucc"+myDKSRef+"\n");
				  correctionOnChangeLeave(pred, failed, myDKSRef);
				  for (Iterator it=backList.listIterator(); it.hasNext();) {
					  DKSRef n = (DKSRef) it.next();
					  send(n, new NodeLeftMsg(failed, frontList, removedNonces.getRecent(NONCESENDSIZE)));
				  }
				  
				  for (Iterator it=frontList.listIterator(); it.hasNext();) {
					  DKSRef n = (DKSRef) it.next();
					  send(n, new NodeLeftMsg(failed, backList, removedNonces.getRecent(NONCESENDSIZE)));
				  }
				  
				  try {
					  final DKSRef pred2 = DKSRef.valueOf(pred.getDKSURL());
					  Runnable callbackObj = new Runnable() {
						  public void run() {
							  appHandler.failCallback(failed, pred2);
						  }
					  };
					  //Thread thread = new Thread(callbackObj);
					  //thread.setName(DKSNode.class.getName() + ".failureHandlerCallback");
					  //thread.start(); // spawn callback in a separate thread
                      threadPool.addJob(callbackObj);
					  
				  } catch (Exception ex) { ex.printStackTrace(); }
			  }
		  }
		  lastTimeoutId = ((HeartbeatMsg) fm.getMsg()).getValue();
	  }
  }

  public void heartbeatH(DKSRef succ, HeartbeatMsg hm) {

  }

  public boolean addMsgHandler(DKSMessage msg, Object handlerObject, String methodName) {
	  DKSOverlayAddress oa = myDKSRef.getOverlayAddress();
	  return marshal.addMsgHandler(oa, msg.getClass().getName(),
			  handlerObject.getClass().getName(), methodName, handlerObject);
  }
  
  protected void installHandlers() {
    DKSOverlayAddress oa = myDKSRef.getOverlayAddress();
    DKSMessage.addMessageTypePrefixed("LOOKUPREQUEST", "dks_marshal.LookupRequestMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.LookupRequestMsg",
				  "dks_node.DKSNode", "lookupRequestH", this);

    DKSMessage.addMessageTypePrefixed("LOOKUP", "dks_marshal.LookupMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.LookupMsg", "dks_node.DKSNode",
				  "lookupH", this);

    DKSMessage.addMessageTypePrefixed("LOOKUPRESPONSE", "dks_marshal.LookupResponseMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.LookupResponseMsg",
				  "dks_node.DKSNode", "lookupResponseH", this);

    DKSMessage.addMessageTypePrefixed("ACKJOININIT", "dks_marshal.AckJoinInitMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.AckJoinInitMsg", "dks_node.DKSNode",
				  "ackJoinInitH", this);

    DKSMessage.addMessageTypePrefixed("HELLO", "dks_marshal.HelloMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.HelloMsg", "dks_node.DKSNode",
				  "helloH", this);

    DKSMessage.addMessageTypePrefixed("ACKHELLO", "dks_marshal.AckHelloMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.AckHelloMsg", "dks_node.DKSNode",
				  "ackHelloH", this);

    DKSMessage.addMessageTypePrefixed("BECOMENORMAL", "dks_marshal.BecomeNormalMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.BecomeNormalMsg", "dks_node.DKSNode",
				  "becomeNormalH", this);

    DKSMessage.addMessageTypePrefixed("JOININIT", "dks_marshal.JoinInitMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.JoinInitMsg", "dks_node.DKSNode",
				  "joinInitH", this);

    DKSMessage.addMessageTypePrefixed("RESTARTJOIN", "dks_marshal.RestartJoinMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.RestartJoinMsg", "dks_node.DKSNode",
				  "restartJoinH", this);

    DKSMessage.addMessageTypePrefixed("ADAPT", "dks_marshal.AdaptMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.AdaptMsg", "dks_node.DKSNode",
				  "adaptH", this);

    DKSMessage.addMessageTypePrefixed("ADAPTLEAVE", "dks_marshal.AdaptLeaveMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.AdaptLeaveMsg", "dks_node.DKSNode",
				  "adaptLeaveH", this);

    DKSMessage.addMessageTypePrefixed("LEAVEREQUEST", "dks_marshal.LeaveRequestMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.LeaveRequestMsg", "dks_node.DKSNode",
				  "leaveRequestH", this);

    DKSMessage.addMessageTypePrefixed("LEAVEREJECT", "dks_marshal.LeaveRejectMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.LeaveRejectMsg", "dks_node.DKSNode",
				  "leaveRejectH", this);

    DKSMessage.addMessageTypePrefixed("LEAVEACCEPT", "dks_marshal.LeaveAcceptMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.LeaveAcceptMsg", "dks_node.DKSNode",
				  "leaveAcceptH", this);

    DKSMessage.addMessageTypePrefixed("BYE", "dks_marshal.ByeMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.ByeMsg", "dks_node.DKSNode",
				  "byeH", this);

    DKSMessage.addMessageTypePrefixed("ACKBYE", "dks_marshal.AckByeMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.AckByeMsg", "dks_node.DKSNode",
				  "ackByeH", this);

    DKSMessage.addMessageTypePrefixed("LOOKUPRESULT", "dks_marshal.LookupResultMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.LookupResultMsg", "dks_node.DKSNode",
				  "lookupResultH", this);

    DKSMessage.addMessageTypePrefixed("INSERTNODE", "dks_marshal.InsertNodeMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.InsertNodeMsg", "dks_node.DKSNode",
				  "insertNodeH", this);

    DKSMessage.addMessageTypePrefixed("BADPOINTER", "dks_marshal.BadPointerMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.BadPointerMsg", "dks_node.DKSNode",
				  "badPointerH", this);

    DKSMessage.addMessageTypePrefixed("BROADCAST", "dks_marshal.BroadCastMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.BroadCastMsg", "dks_node.DKSNode",
				  "broadCastH", this);

    DKSMessage.addMessageTypePrefixed("CORRECTIONONJOIN",
                           "dks_marshal.CorrectionOnJoinMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.CorrectionOnJoinMsg",
				  "dks_node.DKSNode", "cocJoinInterleavedH", this);

    DKSMessage.addMessageTypePrefixed("RESTARTOPERATION",
                           "dks_marshal.RestartOperationMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.RestartOperationMsg",
				  "dks_node.DKSNode", "restartOperationH", this);

    DKSMessage.addMessageTypePrefixed("CORRECTIONONLEAVE",
                           "dks_marshal.CorrectionOnLeaveMsg");

    DKSMessage.addMessageTypePrefixed("NODELEFTMSG",
                           "dks_marshal.NodeLeftMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.NodeLeftMsg",
				  "dks_node.DKSNode", "nodeLeftH", this);

    DKSMessage.addMessageTypePrefixed("FAILUREMSG", "dks_marshal.FailureMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.FailureMsg", "dks_node.DKSNode",
				  "failureHandler", this);

    DKSMessage.addMessageTypePrefixed("HEARTBEATMSG", "dks_marshal.HeartbeatMsg");
    marshal.addMsgHandlerPrefixed(oa, "dks_marshal.HeartbeatMsg", "dks_node.DKSNode",
				  "heartbeatH", this);

  }

} //DKSNode class
